package app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import app.BaseAppV1;
import beans.TrafficHomeDetailPageViewBean;
import common.Constant;
import utils.AtguiguUtil;
import utils.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

public class Dws_03_DwsTrafficPageViewWindow extends BaseAppV1 {
    public static void main(String[] args) {
        new Dws_03_DwsTrafficPageViewWindow().init(4003, 2, "Dws_TrafficPageViewWindow", Constant.TOPIC_DWD_TRAFFIC_PAGE);
    }

    @Override
    protected void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {
        //1.过滤出 course_list 与 home
        SingleOutputStreamOperator<JSONObject> homeAndCourseStream = homeAndCourseStream(stream);

        //2.封装到 pojo 中
        SingleOutputStreamOperator<TrafficHomeDetailPageViewBean> beanStream = pojoStream(homeAndCourseStream);

        //3.开窗 聚合
        SingleOutputStreamOperator<TrafficHomeDetailPageViewBean> resultStream = windowAndAgg(beanStream);
        resultStream.print();

        //4.写出到clickhouse
        writeToClickhouse(resultStream);

    }


    //4.写出到clickhouse
    private void writeToClickhouse(SingleOutputStreamOperator<TrafficHomeDetailPageViewBean> resultStream) {
        resultStream.addSink(FlinkSinkUtil.getClickHouseSink("dws_traffic_page_view_window", TrafficHomeDetailPageViewBean.class));
    }


    //3.开窗 聚合
    private SingleOutputStreamOperator<TrafficHomeDetailPageViewBean> windowAndAgg(SingleOutputStreamOperator<TrafficHomeDetailPageViewBean> beanStream) {
        return beanStream.assignTimestampsAndWatermarks(WatermarkStrategy.<TrafficHomeDetailPageViewBean>forBoundedOutOfOrderness(Duration.ofSeconds(3)).withTimestampAssigner((bean, ts) -> bean.getTs())).windowAll(TumblingEventTimeWindows.of(Time.seconds(5))).reduce(new ReduceFunction<TrafficHomeDetailPageViewBean>() {
            @Override
            public TrafficHomeDetailPageViewBean reduce(TrafficHomeDetailPageViewBean value1, TrafficHomeDetailPageViewBean value2) throws Exception {
                value1.setHomeUvCt(value1.getHomeUvCt() + value2.getHomeUvCt());
                value1.setCourseDetailUvCt(value1.getCourseDetailUvCt() + value2.getCourseDetailUvCt());
                return value1;
            }
        }, new ProcessAllWindowFunction<TrafficHomeDetailPageViewBean, TrafficHomeDetailPageViewBean, TimeWindow>() {
            @Override
            public void process(Context context, Iterable<TrafficHomeDetailPageViewBean> elements, Collector<TrafficHomeDetailPageViewBean> out) throws Exception {
                //获取bean对象
                TrafficHomeDetailPageViewBean bean = elements.iterator().next();
                bean.setStt(AtguiguUtil.toDateTime(context.window().getStart()));
                bean.setEdt(AtguiguUtil.toDateTime(context.window().getEnd()));

                bean.setTs(System.currentTimeMillis());

                out.collect(bean);
            }
        });
    }


    //2.封装到 pojo 中
    private SingleOutputStreamOperator<TrafficHomeDetailPageViewBean> pojoStream(SingleOutputStreamOperator<JSONObject> hAndCStream) {
        return hAndCStream.keyBy(obj -> obj.getJSONObject("common").getString("mid")).process(new KeyedProcessFunction<String, JSONObject, TrafficHomeDetailPageViewBean>() {
            private ValueState<String> homeState; // home页面 状态
            private ValueState<String> courseState; // course_list页面

            @Override
            public void open(Configuration parameters) throws Exception {
                homeState = getRuntimeContext().getState(new ValueStateDescriptor<String>("homeState", String.class));
                courseState = getRuntimeContext().getState(new ValueStateDescriptor<String>("courseState", String.class));
            }

            @Override
            public void processElement(JSONObject value, Context ctx, Collector<TrafficHomeDetailPageViewBean> out) throws Exception {

                String pageId = value.getJSONObject("page").getString("page_id");
                Long ts = value.getLong("ts");
                String today = AtguiguUtil.toDate(ts);

                String homeDate = homeState.value(); //获取homeState状态的值
                String courseDate = courseState.value(); //获取courseState状态的值

                // 首页独立访客数
                Long homeUvCt = 0L;
                if ("home".equals(pageId) && !today.equals(homeDate)) {
                    //更新状态
                    homeState.update(today);
                    homeUvCt = 1L;
                }

                // 课程详情页独立访客数
                Long courseDetailUvCt = 0L;
                if ("course_list".equals(pageId) && !today.equals(courseDate)) {
                    //更新状态
                    courseState.update(today);
                    courseDetailUvCt = 1L;
                }

                TrafficHomeDetailPageViewBean bean = new TrafficHomeDetailPageViewBean("", "", homeUvCt, courseDetailUvCt, ts);
                if (homeUvCt + courseDetailUvCt == 1) {
                    out.collect(bean);
                }
            }
        });
    }


    //1.过滤出 course_list 与 home
    private SingleOutputStreamOperator<JSONObject> homeAndCourseStream(DataStreamSource<String> stream) {
        return stream.map(JSON::parseObject).filter(new FilterFunction<JSONObject>() {
            @Override
            public boolean filter(JSONObject value) throws Exception {
                String pageId = value.getJSONObject("page").getString("page_id");
                return "course_detail".equals(pageId) || "home".equals(pageId);
            }
        });
    }
}
